/*
 * Copyright 2019 gRPC authors.
 *
 * Licensed under the Apache License, Version 2.0 (the "License");
 * you may not use this file except in compliance with the License.
 * You may obtain a copy of the License at
 *
 *     http://www.apache.org/licenses/LICENSE-2.0
 *
 * Unless required by applicable law or agreed to in writing, software
 * distributed under the License is distributed on an "AS IS" BASIS,
 * WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
 * See the License for the specific language governing permissions and
 * limitations under the License.
 *
 */

// Package resolver implements the xds resolver, that does LDS and RDS to find
// the cluster to use.
package resolver

import (
	"encoding/json"
	"errors"
	"fmt"
	"github.com/eapache/channels"
	v2corepb "github.com/envoyproxy/go-control-plane/envoy/api/v2/core"
	v3corepb "github.com/envoyproxy/go-control-plane/envoy/config/core/v3"
	"google.golang.org/grpc/credentials"
	"google.golang.org/protobuf/types/known/structpb"
	"tmlake.com/api-gateway/istio-agent/pkg/grpclog/inside"
	"tmlake.com/api-gateway/istio-agent/pkg/grpcsync"
	"tmlake.com/api-gateway/istio-agent/pkg/model"
	"tmlake.com/api-gateway/istio-agent/pkg/resolver"
	"tmlake.com/api-gateway/istio-agent/xds/xdsclient"
	"tmlake.com/api-gateway/istio-agent/xds/xdsclient/xdsresource/version"

	"tmlake.com/api-gateway/istio-agent/xds/xdsclient/xdsresource"
)

const xdsScheme = "xds"

// For overriding in unittests.
var newXDSClient = func() (xdsclient.XDSClient, error) { return xdsclient.New() }

func initXDSClient() (xdsclient.XDSClient, error) {
	newXDSClient := newXDSClient

	client, err := newXDSClient()
	if err != nil {
		return nil, fmt.Errorf("xds: failed to create xds-client: %v", err)
	}

	fmt.Printf("Created an xdsClient\n")
	return client, nil
}

func init() {
	resolver.Register(&xdsResolverBuilder{})
}

type xdsResolverBuilder struct {
	newXDSClient func() (xdsclient.XDSClient, error)
}

func (b *xdsResolverBuilder) Build(client xdsclient.XDSClient, opts resolver.BuildOptions) (_ resolver.Resolver, retErr error) {
	r := &XdsResolver{
		Client:      client,
		closed:      grpcsync.NewEvent(),
		rdsUpdateCh: make(chan rdsHandlerUpdate, 1),
		ldsUpdateCh: make(chan ldsUpdateWithError, 1),
		edsUpdateCh: make(chan edsHandlerUpdate, 1),
		cdsUpdateCh: make(chan cdsUpdateWithError, 1),
		listenerWrapper: struct {
			listener map[string]*rdsHandler
		}{
			listener: make(map[string]*rdsHandler),
		},
		clusterWrapper: struct {
			cluster map[string]*edsHandler
		}{
			cluster: make(map[string]*edsHandler),
		},
		updateCh: opts.UpdateCh,
		node:     opts.Node,
	}

	defer func() {
		if retErr != nil {
			r.Close()
		}
	}()
	r.logger = prefixLogger(r)
	r.logger.Infof("Creating resolver")

	bootstrapConfig := client.BootstrapConfig()
	if bootstrapConfig == nil {
		return nil, errors.New("bootstrap configuration is empty")
	}

	switch bootstrapConfig.XDSServer.TransportAPI {
	case version.TransportV2:
		v2, _ := bootstrapConfig.XDSServer.NodeProto.(*v2corepb.Node)
		if v2 != nil {
			v2.Id = opts.Node.ID
			r.logger.Errorf("V2 nodeId: %s", v2.Id)
			xdsMeta, err := b.extractMeta(opts.Node)
			if err != nil {
				return nil, fmt.Errorf("failed extracting xds metadata: %v", err)
			}
			v2.Metadata = xdsMeta
			bytes, err := json.Marshal(opts.Node.Metadata)
			if err == nil {
				r.logger.Errorf("V2 nodeMetadata: %s", string(bytes))
			}
		}
	case version.TransportV3:
		v3, _ := bootstrapConfig.XDSServer.NodeProto.(*v3corepb.Node)
		if v3 != nil {
			v3.Id = opts.Node.ID
			r.logger.Errorf("V3 nodeId: %s", v3.Id)
			xdsMeta, err := b.extractMeta(opts.Node)
			if err != nil {
				return nil, fmt.Errorf("failed extracting xds metadata: %v", err)
			}
			v3.Metadata = xdsMeta
			bytes, err := json.Marshal(opts.Node.Metadata)
			if err == nil {
				r.logger.Errorf("V3 nodeMetadata: %s", string(bytes))
			}
		}

	}

	for _, a := range bootstrapConfig.Authorities {
		if a.XDSServer == nil {
			continue
		}
		switch a.XDSServer.TransportAPI {
		case version.TransportV2:
			v2, _ := a.XDSServer.NodeProto.(*v2corepb.Node)
			if v2 != nil {
				v2.Id = opts.Node.ID
				r.logger.Errorf("V2 nodeId: %s", v2.Id)
				xdsMeta, err := b.extractMeta(opts.Node)
				if err != nil {
					return nil, fmt.Errorf("failed extracting xds metadata: %v", err)
				}
				v2.Metadata = xdsMeta
				bytes, err := json.Marshal(opts.Node.Metadata)
				if err == nil {
					r.logger.Errorf("V2 nodeMetadata: %s", string(bytes))
				}
			}
		case version.TransportV3:
			v3, _ := a.XDSServer.NodeProto.(*v3corepb.Node)
			if v3 != nil {
				v3.Id = opts.Node.ID
				r.logger.Errorf("V3 nodeId: %s", v3.Id)

				xdsMeta, err := b.extractMeta(opts.Node)
				if err != nil {
					return nil, fmt.Errorf("failed extracting xds metadata: %v", err)
				}
				v3.Metadata = xdsMeta
				bytes, err := json.Marshal(opts.Node.Metadata)
				if err == nil {
					r.logger.Errorf("V3 nodeMetadata: %s", string(bytes))
				}

			}
		}
	}

	// If xds credentials were specified by the user, but bootstrap configs do
	// not contain any certificate provider configuration, it is better to fail
	// right now rather than failing when attempting to create certificate
	// providers after receiving an CDS response with security configuration.
	var creds credentials.TransportCredentials
	switch {
	case opts.DialCreds != nil:
		creds = opts.DialCreds
	case opts.CredsBundle != nil:
		creds = opts.CredsBundle.TransportCredentials()
	}
	if xc, ok := creds.(interface{ UsesXDS() bool }); ok && xc.UsesXDS() {
		if len(bootstrapConfig.CertProviderConfigs) == 0 {
			return nil, errors.New("xds: xdsCreds specified but certificate_providers config missing in bootstrap file")
		}
	}

	cancelClusterWatch := r.Client.WatchCluster(xdsresource.AllClusterResource, r.handleClusterUpdate)
	cancelListenerWatch := r.Client.WatchListener(xdsresource.AllListenerResource, r.handleListenerUpdate)

	r.logger.Infof("Watch started on resource name [Cluster:\"%s\", Listener:\"%s\"] with xds-client %p", xdsresource.AllListenerResource, xdsresource.AllClusterResource, r.Client)
	r.cancelWatch = func() {
		cancelListenerWatch()
		cancelClusterWatch()
		r.logger.Infof("Watch cancel on resource name [Cluster:\"%s\", Listener:\"%s\"] with xds-client %p", xdsresource.AllListenerResource, xdsresource.AllClusterResource, r.Client)
	}

	go r.run()
	return r, nil

}

// Name helps implement the resolver.Builder interface.
func (*xdsResolverBuilder) Scheme() string {
	return xdsScheme
}

func (*xdsResolverBuilder) extractMeta(node *model.Node) (*structpb.Struct, error) {
	bytes, err := json.Marshal(node.Metadata)
	if err != nil {
		return nil, err
	}
	rawMeta := map[string]interface{}{}
	if err := json.Unmarshal(bytes, &rawMeta); err != nil {
		return nil, err
	}
	xdsMeta, err := structpb.NewStruct(rawMeta)
	if err != nil {
		return nil, err
	}
	return xdsMeta, nil
}

// EventType type of event associated with an informer
type EventType string

const (
	// CreateEvent event associated with new objects in an informer
	CreateEvent EventType = "CREATE"
	// UpdateEvent event associated with an object update in an informer
	UpdateEvent EventType = "UPDATE"
	// DeleteEvent event associated when an object is removed from an informer
	DeleteEvent EventType = "DELETE"
	// ConfigurationEvent event associated when a controller configuration object is created or updated
	ConfigurationEvent EventType = "CONFIGURATION"
)

// Event holds the context of an event.
type Event struct {
	Type EventType
	Obj  interface{}
}

type ldsUpdateWithError struct {
	update xdsresource.ListenerUpdate
	err    error
}

type cdsUpdateWithError struct {
	update xdsresource.ClusterUpdate
	err    error
}

func New(updateCh *channels.RingChannel, node *model.Node) (_ resolver.Resolver, retErr error) {
	client, err := initXDSClient()
	if err != nil {
		return nil, err
	}

	b := resolver.Get("xds")
	if b == nil {
		return nil, errors.New("scheme is not recognized")
	}

	opts := resolver.BuildOptions{
		UpdateCh: updateCh,
		Node:     node,
	}
	r, err := b.Build(client, opts)
	if err != nil {
		return nil, err
	}

	return r, nil
}

// XdsResolver implements the resolver.Resolver interface.
//
// It registers a watcher for ServiceConfig updates with the xdsClient object
// (which performs LDS/RDS queries for the same), and passes the received
// updates to the ClientConn.
type XdsResolver struct {
	//target resolver.Target
	//cc     resolver.ClientConn
	closed *grpcsync.Event

	logger *grpclog.PrefixLogger

	// The underlying xdsClient which performs all xDS requests and responses.
	Client xdsclient.XDSClient

	// cancelWatch is the function to cancel the watcher.
	cancelWatch func()

	clusterWrapper  ClusterWrapper
	listenerWrapper ListenerWrapper
	// ldsUpdateCh is a channel for XDSClient LDS updates.
	ldsUpdateCh chan ldsUpdateWithError
	// rdsUpdateCh is a channel for XDSClient RDS updates.
	rdsUpdateCh chan rdsHandlerUpdate
	// cdsUpdateCh is a channel for XDSClient CDS updates.
	cdsUpdateCh chan cdsUpdateWithError
	// rdsUpdateCh is a channel for XDSClient RDS updates.
	edsUpdateCh chan edsHandlerUpdate

	updateCh *channels.RingChannel
	node     *model.Node
}

func (r *XdsResolver) handleListenerUpdate(update xdsresource.ListenerUpdate, err error) {
	r.ldsUpdateCh <- ldsUpdateWithError{update: update, err: err}
}

// handleClusterUpdate is the callback which handles LDS Updates. It writes the
// received update to the update channel, which is picked up by the run
// goroutine.
func (r *XdsResolver) handleClusterUpdate(update xdsresource.ClusterUpdate, err error) {
	r.cdsUpdateCh <- cdsUpdateWithError{update: update, err: err}
}

// run is a long running goroutine which blocks on receiving service updates
// and passes it on the ClientConn.
func (r *XdsResolver) run() {
	for {
		select {
		case <-r.closed.Done():
			return
		case u := <-r.ldsUpdateCh:
			r.handleLDSUpdate(u)
		case u := <-r.rdsUpdateCh:
			r.handleRDSUpdate(u)
		case u := <-r.cdsUpdateCh:
			r.handleCDSUpdate(u)
		case u := <-r.edsUpdateCh:
			r.handleEDSUpdate(u)
		}
	}
}

func (r *XdsResolver) handleLDSUpdate(update ldsUpdateWithError) {
	r.logger.Infof("begin handle LDS: '%s' ...\n", update.update.ListenerName)
	if update.err != nil {
		r.logger.Warningf("Received error for resource %q: %+v", update.update.ListenerName, update.err)
		if xdsresource.ErrType(update.err) == xdsresource.ErrorTypeResourceNotFound {
			if update.update.ListenerName != "" && update.update.ListenerName != xdsresource.AllListenerResource {
				rdsHandler := r.listenerWrapper.listener[update.update.ListenerName]
				rdsHandler.close()
				delete(r.listenerWrapper.listener, update.update.ListenerName)
			}

		}
		// For errors which are anything other than "resource-not-found", we
		// continue to use the old configuration.
		return
	}

	if update.update.ListenerName != "" && update.update.ListenerName != xdsresource.InboundListenerName {
		if r.listenerWrapper.listener[update.update.ListenerName] == nil {
			r.listenerWrapper.listener[update.update.ListenerName] = newRDSHandler(r.Client, r.rdsUpdateCh, update.update, r.logger)
		}

		//Download route configuration resources through RDS
		r.listenerWrapper.listener[update.update.ListenerName].updateRouteNamesToWatch(update.update.InboundListenerCfg.FilterChains.RouteConfigNames)
	}

	if update.update.ListenerName == xdsresource.InboundListenerName && update.update.InboundListenerCfg.Address == xdsresource.InboundListenerAddress &&
		update.update.InboundListenerCfg.Port == xdsresource.InboundListenerPort {
		//Convert to njet configuration
		r.logger.Infof("'%s' LDS begins conversion to njet configuration...\n", xdsresource.InboundListenerName)

		//create njet inbound traffic hijacking module
		r.updateCh.In() <- Event{
			Type: UpdateEvent,
			Obj:  update.update.InboundListenerCfg,
		}
	}

}

// handleRDSUpdate handles a full rds update from rds handler. On a successful
// update, the server will switch to ServingModeServing as the full
// configuration (both LDS and RDS) has been received.
func (r *XdsResolver) handleRDSUpdate(update rdsHandlerUpdate) {
	if update.err != nil {
		r.logger.Warningf("Received error for rds names, err: %+v", update.err)

		// For errors which are anything, we
		// continue to use the old configuration.
		return
	}

	//Convert to njet configuration
	r.logger.Infof("RouteConfigUpdate begins conversion to njet configuration...")

	//create njet server module
	r.updateCh.In() <- Event{
		Type: UpdateEvent,
		Obj:  update,
	}
}
func (r *XdsResolver) handleCDSUpdate(update cdsUpdateWithError) {
	if update.err != nil {
		r.logger.Warningf("Received error for resource %q: %+v", update.update.ClusterName, update.err)
		if xdsresource.ErrType(update.err) == xdsresource.ErrorTypeResourceNotFound {
			if update.update.ClusterName != "" && update.update.ClusterName != xdsresource.AllClusterResource {
				rdsHandler := r.clusterWrapper.cluster[update.update.ClusterName]
				rdsHandler.close()
				delete(r.clusterWrapper.cluster, update.update.ClusterName)
			}

		}
		// For errors which are anything other than "resource-not-found", we
		// continue to use the old configuration.
		return
	}

	if update.update.ClusterName != "" {
		if r.clusterWrapper.cluster[update.update.ClusterName] == nil {
			r.clusterWrapper.cluster[update.update.ClusterName] = newEDSHandler(r.Client, r.edsUpdateCh, update.update, r.logger)
		}

		r.logger.Debugf("EDSServiceName:%s", update.update.EDSServiceName)
		//Download endpoint configuration resources through EDS
		r.clusterWrapper.cluster[update.update.ClusterName].updateEndpointNamesToWatch(update.update.EDSServiceName)

		//Convert to njet configuration
		r.logger.Infof("ClusterUpdate begins conversion to njet configuration...")

		//create upstream ssl etc.
		r.updateCh.In() <- Event{
			Type: UpdateEvent,
			Obj:  update,
		}
	}
}
func (r *XdsResolver) handleEDSUpdate(update edsHandlerUpdate) {
	if update.err != nil {
		r.logger.Warningf("Received error for eds names, err: %+v", update.err)

		// For errors which are anything, we
		// continue to use the old configuration.
		return
	}

	//Convert to njet configuration
	r.logger.Infof("handleEDSUpdate EndpointUpdate begins conversion to njet configuration...")

	////create njet upstream module
	r.updateCh.In() <- Event{
		Type: UpdateEvent,
		Obj:  update,
	}
}

// ResolveNow is a no-op at this point.
func (*XdsResolver) ResolveNow(o resolver.ResolveNowOptions) {}

// Close closes the resolver, and also closes the underlying xdsClient.
func (r *XdsResolver) Close() {
	// Note that Close needs to check for nils even if some of them are always
	// set in the constructor. This is because the constructor defers Close() in
	// error cases, and the fields might not be set when the error happens.
	if r.cancelWatch != nil {
		r.cancelWatch()
	}

	for _, rdsHandler := range r.listenerWrapper.listener {
		rdsHandler.close()
	}

	if r.Client != nil {
		r.Client.Close()
	}
	r.closed.Fire()
	r.logger.Infof("Shutdown")
}
